package com.sunday.common.mq.rabbit.study.base.e2_WorkQueues;

import com.rabbitmq.client.*;
import com.sunday.common.mq.rabbit.study.base.ConnectionUtil;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.util.concurrent.TimeUnit;

/**
 * 工作队列
 * <p>
 * 生产者 -> 队列 -> 消耗：用于在多个工作人员之间分配耗时任务的工作队列。
 * 在第一个教程中，我们 编写程序以从命名队列发送和接收消息。在此 我们将创建一个将用于分发的工作队列 多个工作人员之间的耗时任务。
 * <p>
 * 工作队列（又名：任务队列）背后的主要思想是避免 立即执行资源密集型任务，必须等待 它要完成。相反，我们将任务安排在以后完成。我们将任务封装为消息并将其发送到队列。正在运行的工作进程 在后台将弹出任务并最终执行 工作。当您运行许多工作线程时，任务将在它们之间共享。
 * <p>
 * 这个概念在 Web 应用程序中特别有用，因为它 在短 HTTP 请求期间无法处理复杂任务 窗。
 */
@Slf4j
public class Worker {

    private static final String TASK_QUEUE_NAME = "task_queue";

    public static void main(String[] argv) throws Exception {

        ConnectionFactory factory = ConnectionUtil.factory();

        final Connection connection = factory.newConnection();
        final Channel channel = connection.createChannel();

        channel.queueDeclare(TASK_QUEUE_NAME, false, false, false, null);
        log.info(" [*] Waiting for messages. To exit press CTRL+C");

        /**
         * QoS 预取设置对使用 basic.get（“pull API”）提取的消息没有影响，即使在手动确认模式下也是如此。
         *
         * 这是通过使用basic.qos方法设置“预取计数”值  完成的。该值定义了通道上允许的最大未确认递送数量。
         * 一旦数字达到配置的计数，RabbitMQ将停止在通道上传送更多消息，除非至少有一个未确认的消息被确认
         *
         * 预取值 1 是最保守的。它将 显著降低吞吐量，特别是在 使用者连接延迟较高的环境。
         * 为 许多应用程序，较高的值将是合适的，并且 最佳的。
         */
        channel.basicQos(1);

        /**
         * 在rabbitmq中有两种消息处理的模式，一种是推模式/订阅模式/投递模式（也叫push模式），消费者调用channel.basicConsume方法订阅队列后，
         * 由RabbitMQ主动将消息推送给订阅队列的消费者；另一种是拉模式/检索模式（也叫pull模式），需要消费者调用channel.basicGet方法，主动从指定队列中拉取消息。
         *
         * 推模式：消息中间件主动将消息推送给消费者
         * 拉模式：消费者主动从消息中间件拉取消息
         *
         * 推模式（push）
         * 推模式接收消息是最有效的一种消息处理方式。channel.basicConsume(queneName,consumer)方法将信道(channel)设置成投递模式，直到取消队列的订阅为止；在投递模式期间，当消息到达RabbitMQ时，RabbitMQ会自动地、不断地投递消息给匹配的消费者，而不需要消费端手动来拉取，当然投递消息的个数还是会受到channel.basicQos的限制。
         * 推模式将消息提前推送给消费者，消费者必须设置一个缓冲区缓存这些消息。优点是消费者总是有一堆在内存中待处理的消息，所以当真正去消费消息时效率很高。缺点就是缓冲区可能会溢出。
         * 由于推模式是信息到达RabbitMQ后，就会立即被投递给匹配的消费者，所以实时性非常好，消费者能及时得到最新的消息。
         *
         * 拉模式（pull）
         * 如果只想从队列中获取单条消息而不是持续订阅，则可以使用channel.basicGet方法来进行消费消息。
         * 拉模式在消费者需要时才去消息中间件拉取消息，这段网络开销会明显增加消息延迟，降低系统吞吐量。
         * 由于拉模式需要消费者手动去RabbitMQ中拉取消息，所以实时性较差；消费者难以获取实时消息，具体什么时候能拿到新消息完全取决于消费者什么时候去拉取消息。
         */
        push(channel, "A");
        push(channel, "B");

        pull(channel, "C");
        pull(channel, "D");

    }

    /**
     * https://www.rabbitmq.com/api-guide.html#getting
     * 检索单个消息（“pull API”）
     * <p>
     * 也可以按需检索单个消息（“pull API”，即轮询）。 这种消费方法效率非常低，因为它实际上是轮询 并且应用程序必须反复要求结果，即使绝大多数请求 没有结果。因此，强烈建议不要使用此方法。
     * 若要“拉取”消息，请使用 Channel.basicGet 方法。返回值为 GetResponse 实例，标头信息（属性）来自该实例 并且可以提取消息正文：
     * <p>
     * SP: 有很多环境中是需要拉模式的，例如控制并发，控制发送速率等情况
     */
    private static void pull(Channel channel, String mark) {

        new Thread(() -> {

            try {
                while (true) {
                    GetResponse response = channel.basicGet(TASK_QUEUE_NAME, false);
                    if (response == null) {
                        TimeUnit.SECONDS.sleep(1);
                        continue;
                    }
                    String message = new String(response.getBody(), "UTF-8");
                    log.info(" [x] mark '{}'", mark);
                    log.info(" [x] Received '{}'", message);
                    try {
                        doWork(message);
                    } finally {
                        log.info(" [x] Done");
                        channel.basicAck(response.getEnvelope().getDeliveryTag(), false);
                    }
                }
            } catch (Exception e) {
                log.error(e.toString());
            }
        }).start();

    }

    /**
     * https://www.rabbitmq.com/api-guide.html#getting
     * 通过订阅接收消息（“push API”）
     * <p>
     * 接收消息的最有效方法是设置 使用使用者界面进行订阅。然后消息将被传递 在他们到达时自动，而不必 明确要求。
     * 调用与使用者相关的 API 方法时，单个订阅是 总是由他们的消费者标签引用。消费者标签是消费者 标识符，可以是客户端生成的，也可以是服务器生成的。要让 RabbitMQ 生成节点范围的唯一标签，使用 Channel#basicConsumption 覆盖 不接受消费者标签参数或传递空字符串 ，并使用 Channel#basicConsumption 返回的值。 消费者标签用于取消消费者。
     * 不同的使用者实例必须具有不同的 消费者标记。连接上的重复使用者标记是 强烈建议不要使用，并可能导致自动问题 连接恢复和混淆监控数据时 消费者受到监控。
     * 实现消费者的最简单方法是 子类 便利类 DefaultConsumer。 可以通过 basicConsumption 调用传递此子类的对象以设置订阅：
     * <p>
     * 公平调度
     * 您可能已经注意到调度仍然无法正常工作 如我们所愿。例如，在有两个工人的情况下，当所有 奇数消息很重，偶数消息很轻，一个工作人员将是 经常忙碌，另一个几乎不会做任何工作。井 RabbitMQ对此一无所知，仍然会调度 消息均匀。
     * 发生这种情况是因为 RabbitMQ 只是在消息时调度消息 进入队列。它不看未确认的数量 面向消费者的消息。它只是盲目地发送每 n 条消息 给第 n 个消费者。
     * Producer -> Queue -> Consuming: RabbitMQ dispatching messages.
     * 为了解决这个问题，我们可以将 basicQos 方法与预取计数 = 1 设置一起使用。这告诉 RabbitMQ 不要付出更多 一次给工作人员一条消息。或者，换句话说，不要调度 给工作人员的新消息，直到它处理并确认 前一个。相反，它会将其调度给下一个尚未繁忙的工作人员。
     */
    private static void push(Channel channel, String mark) throws IOException {

        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            String message = new String(delivery.getBody(), "UTF-8");
            log.info(" [x] mark '{}'", mark);
            log.info(" [x] Received '{}'", message);
            try {
                doWork(message);
            } finally {
                log.info(" [x] Done");
                /**
                 * 当使用手动确认时，在消息传递和确认之间，到RabbitMQ节点的网络连接可能会失败。
                 * 连接恢复后，RabbitMQ将重置所有通道上的交付标签。这意味着basic.ack，basic.
                 * nack和basic.reject 与旧的交付标签将导致通道异常。 为了避免这种情况，RabbitMQ
                 * Java客户端跟踪并更新交付标签，使它们在恢复之间单调增长。
                 *
                 * Channel.basicAck （basic.ack 用于肯定确认） Channel.basicNack
                 * （basic.nack 用于否定确认（注意：这是对AMQP
                 * 0-9-1的RabbitMQ扩展）nack是reject的升级版，将支持部分打回和全部打回的功能）
                 * Channel.basicReject （basic.reject 用于否定确认）
                 *
                 * 然后将调整后的交付标签转换为RabbitMQ使用的标签。带有陈旧交付标签的确认将不会发送。
                 * 使用手动确认和自动恢复的应用程序必须能够处理重新投递。
                 *
                 */
                channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);

                /**
                 *
                 * 交货的负面确认和重新排列
                 *
                 * 拒绝一个或多个接收到的消息。 从@link
                 * com.rabbitmq.client.AMQP.Basic.GetOk提供“交付标签”。 或{ @link
                 * com.rabbitmq.client.AMQP.Basic。GetOk方法包含要拒绝的消息。
                 *
                 * @see AMQP.Basic.Nack
                 * @param deliveryTag {@link com.rabbitmq.client.AMQP.Basic。GetOk}或{@link com.rabbitmq.client.AMQP.Basic.Deliver}
                 * @param multiple True表示拒绝以下所有消息 所提供的交货标签;False表示只拒绝所提供的 发货标签。
                 * @param requeue 如果拒绝的消息应该重新请求，则为True 比丢弃/死信
                 * @throws io .如果遇到错误，IOException
                 *
                 * void basicNack(long deliveryTag, boolean multiple, boolean requeue) throws IOException;
                 */
//                channel.basicNack(delivery.getEnvelope().getDeliveryTag(), false, false);
//                channel.basicNack(delivery.getEnvelope().getDeliveryTag(), false, true);

            }
        };

        channel.basicConsume(TASK_QUEUE_NAME, false, deliverCallback, consumerTag -> {
        });
    }

    private static void doWork(String task) {
        for (char ch : task.toCharArray()) {
            if (ch == '.') {
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException _ignored) {
                    Thread.currentThread().interrupt();
                }
            }
        }
    }
}

